package net.bwie.vehicle.dws.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;


import net.bwie.vehicle.dws.bean.BatteryHealthData;
import net.bwie.vehicle.dws.function.BatteryHealthWindowFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

/**
 * 窗口内车辆电池健康度统计
 */
public class vehicleWindowBatteryHealthJob {
    public static void main(String[] args) throws Exception {
        // 获取当前执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度为1，确保数据处理的顺序性
        env.setParallelism(1);

        // 从Kafka中消费车辆电池数据，创建数据流
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle_battery_data");

        // 调用BatteryHealth函数处理数据流，评估电池健康状况
        BatteryHealth(kafkaStream);

        // 启动环境执行流处理作业，作业名称为"vehicleWindowBatteryHealth"
        env.execute("vehicleWindowBatteryHealth");

    }

    /**
     * 处理电池健康数据流
     * 该方法从Kafka数据流中读取数据，将其解析为BatteryHealthData对象，并分配时间戳和水印
     *
     * @param kafkaStream 来自Kafka的主题数据流，包含电池健康数据的JSON字符串
     */
    private static void BatteryHealth(DataStream<String> kafkaStream) {
        // 将Kafka数据流中的每条记录解析为BatteryHealthData对象
        SingleOutputStreamOperator<BatteryHealthData> mapStream = kafkaStream.map(o -> JSON.parseObject(o, BatteryHealthData.class));

        // 为BatteryHealthData对象分配时间戳和水印
        // 使用有界无序性策略，允许最大30秒的乱序事件时间
        // 应用场景：处理乱序事件，确保事件时间的正确性
        SingleOutputStreamOperator<BatteryHealthData> batteryHealthDataSingleOutputStreamOperator = mapStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<BatteryHealthData>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withTimestampAssigner(
                        new SerializableTimestampAssigner<BatteryHealthData>() {
                            @Override
                            public long extractTimestamp(BatteryHealthData element, long recordTimestamp) {
                                // 从BatteryHealthData对象中提取时间戳
                                return element.getTimestamp();
                            }
                        }
                )
        );

// 将电池健康数据按照所有记录分组，并定义一个翻滚事件时间窗口，窗口大小为5分钟
        WindowedStream<BatteryHealthData, String, TimeWindow> window = batteryHealthDataSingleOutputStreamOperator.keyBy(o -> "all").window(TumblingEventTimeWindows.of(Time.minutes(5)));

// 处理窗口中的数据，应用自定义的窗口函数以计算电池健康指标
        SingleOutputStreamOperator<String> apply = window.apply(new BatteryHealthWindowFunction());

// 将计算结果 sinking 到 ClickHouse 数据库中，采用 upsert 方式更新或插入数据
// 这里使用了自定义的 JdbcUtil 工具类来实现数据 sinking
        JdbcUtil.sinkToClickhouseUpsert(apply,
                "INSERT INTO new_car.vehicle_dws_battery_Health(\n" +
                        "startTime, endTime, vehicleCount, sumTemp, avgTemp, ts)\n" +
                        "VALUES(?,?,?,?,?,?)");

    }
}
